# _*_ coding:utf-8

from kafka import KafkaConsumer
import time
import json


consumer = KafkaConsumer(
    bootstrap_servers = ["172.16.197.88:9092","172.16.197.90:9092","172.16.197.89:9092"],
    auto_offset_reset = 'earliest'
    )

TopicPartition = namedtuple("TopicPartition", ["topic", "partition"])
consumer.assign([TopicPartition(topic = 'gps_info0', partition=1)])


while True:
    #返回值是一个字典(字典的key为各个分区, 字段的value为ConsumerRecord类型的list)
    print('start one poll')
    msg = consumer.poll(timeout_ms=5)  
    #遍历字典    
    for key in msg.keys():
        tmp = list(msg.get(key))
        if tmp is not None:
            i = 0
            while i < len(tmp):
                str = "value is {value}, partition is {partition}".format(value=tmp[i].value, partition=tmp[i].partition)
                print(str)
                i = i + 1
    time.sleep(2)